package com.bfxy.rabbit.producer.task;

import com.bfxy.rabbit.producer.broker.RabbitBroker;
import com.bfxy.rabbit.producer.constant.BrokerMessageStatus;
import com.bfxy.rabbit.producer.entity.BrokerMessage;
import com.bfxy.rabbit.producer.service.MessageStoreService;
import com.bfxy.rabbit.task.annotation.ElasticJobConfig;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @describte: 数据可靠性投递的补偿机制
 * @program:rabbit-parent
 * @author:zhengdp
 * @create:2021-09-16 17:40:15
 **/
@Component
@ElasticJobConfig(
        name="com.bfxy.rabbit.producer.task.RetryMessageDataflowJob",
        cron = "0/10 * * * * ?",
        description = "可靠性投递消息补偿任务",
        overwrite = true,
        shardingTotalCount = 1
)
@Slf4j
public class RetryMessageDataflowJob implements DataflowJob<BrokerMessage> {

    @Autowired
    private MessageStoreService messageStoreService;

    @Autowired
    private RabbitBroker rabbitBroker;

    /*
    最大重试次数
     */
    private static final int MAX_RETRY_COUNT = 3;

    /*
    从数据库中抓取到 未被确认且超时的任务 （使用Service从数据库中抓取数据）
     */
    @Override
    public List<BrokerMessage> fetchData(ShardingContext shardingContext) {
        List<BrokerMessage> list = messageStoreService.fetchTimeOutMeesage4Retry(BrokerMessageStatus.SENDING);
        log.info("---------@@@@ 抓取数据集合，数量： {} @@@@----------",list.size());
        return list;
    }

    /*
    重新发送消息 (使用Broker来进行消息的发送）
     */
    @Override
    public void processData(ShardingContext shardingContext, List<BrokerMessage> data) {
        // 利用 rabbitBroker 重新发送消息
        data.forEach(brokerMessage -> {
            String messageId = brokerMessage.getMessageId();
            // 判断重试次数是否超过最大重试次数，如果超过，就直接将日志状态设置为发送失败
            if(brokerMessage.getTryCount() >= MAX_RETRY_COUNT){
                this.messageStoreService.failure(messageId);
                log.warn(" -----消息设置为最终失败，消息ID: {} -------", messageId);
            }else{
                // 未超过，重新发送消息
                // 先更新一下 tryCount字段，其实我感觉 nextTry字段也是需要更新的（也就是下次重试时间）
                this.messageStoreService.updateTryCount(messageId);
                // 利用 broker重新发送消息
                rabbitBroker.reliantSend(brokerMessage.getMessage());
            }
        });
    }
}
